package org.jetlinks.pro.device.service;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.assets.*;
import org.jetlinks.pro.assets.event.AssetsBindEvent;
import org.jetlinks.pro.assets.event.AssetsUnBindAllEvent;
import org.jetlinks.pro.assets.event.AssetsUnBindEvent;
import org.jetlinks.pro.assets.impl.terms.DimensionAssetsTerm;
import org.jetlinks.pro.assets.impl.terms.DimensionAssetsTermBuilder;
import org.jetlinks.pro.device.entity.DeviceInstanceEntity;
import org.jetlinks.pro.device.events.DeviceDeployedEvent;
import org.jetlinks.pro.device.tenant.DeviceAssetType;
import org.jetlinks.pro.tenant.TenantAssetManager;
import org.jetlinks.pro.tenant.TenantAsset;
import org.jetlinks.pro.tenant.event.TenantAssetsBindEvent;
import org.jetlinks.pro.tenant.event.TenantAssetsUnBindEvent;
import org.jetlinks.pro.tenant.term.AssetsTerm;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 设备和租户关联同步器,用于设备关联了租户时,将租户信息同步到设备配置中,
 * 以便在进行消息转发时,可获取设备消息的租户信息
 *
 * @author zhouhao
 * @see org.jetlinks.pro.device.message.DeviceMessageConnector
 * @since 1.6
 */
@Component
@AllArgsConstructor
@Slf4j
public class DeviceTenantSynchronizer {

    private final DeviceRegistry deviceRegistry;

    private final AssetBindManager bindManager;

    private final TenantAssetManager assetManager;

    private final LocalDeviceInstanceService instanceService;

    private final EmitterProcessor<String> deviceIdProcessor = EmitterProcessor.create(false);

    private final FluxSink<String> deviceIdSink = deviceIdProcessor.sink(FluxSink.OverflowStrategy.BUFFER);

    @PostConstruct
    public void init() {

        //用时间缓存去重，优化批量处理时可能对同一个设备进行重复的操作
        deviceIdProcessor
            .buffer(Duration.ofMillis(500))
            .flatMap(list -> this
                .syncDeviceBinding(new HashSet<>(list))
                .onErrorResume(err -> {
                    log.error(err.getMessage(), err);
                    return Mono.empty();
                }))
            .subscribe();

    }

    /**
     * 监听租户资产绑定事件
     *
     * @param event 事件
     */
    @EventListener
    public void handleBindEvent(TenantAssetsBindEvent event) {
        if (DeviceAssetType.device.getId().equals(event.getAssetType())) {
            event.getAssetId().forEach(deviceIdSink::next);
        }
    }

    @EventListener
    public void handleUnBindEvent(AssetsBindEvent event) {
        if (DeviceAssetType.device.getId().equals(event.getAssetType())) {
            event.getAssetId().forEach(deviceIdSink::next);
        }
    }

    @EventListener
    @SuppressWarnings("all")
    public void handleUnBindEvent(AssetsUnBindEvent event) {
        if (DeviceAssetType.device.getId().equals(event.getAssetType())) {
            event.getAssetId().forEach(deviceIdSink::next);
        }
    }

    @EventListener
    @SuppressWarnings("all")
    public void handleUnBindEvent(TenantAssetsUnBindEvent event) {
        if (DeviceAssetType.device.getId().equals(event.getAssetType())) {
            event.getAssetId().forEach(deviceIdSink::next);
        }
    }

    @EventListener
    public void handleUnBindAllEvent(AssetsUnBindAllEvent event) {
        if (DeviceAssetType.device.getId().equals(event.getAssetType())) {
            event.async(
                syncDeviceBinding(event.getAssetId())
            );
        }
    }

    @EventListener
    public void handleDeviceDeploy(DeviceDeployedEvent event) {
        event.async(
            syncDeviceBinding(
                event.getDevices().stream().map(DeviceInstanceEntity::getId).collect(Collectors.toList())
            ));
    }

    public Mono<Void> syncDeviceBinding(Collection<String> deviceIds) {
        Set<String> noBindDevice = new HashSet<>(deviceIds);

        return bindManager
            .getBindings(DeviceAssetType.device, deviceIds)
            .groupBy(AssetBinding::getAssetId, Integer.MAX_VALUE)
            .flatMap(group -> {
                String deviceId = group.key();
                noBindDevice.remove(deviceId);
                Flux<AssetBinding> cache = group.cache();
                return Mono
                    .zip(
                        //设备
                        deviceRegistry.getDevice(deviceId),
                        //租户
                        cache.filter(binding -> binding instanceof TenantAsset)
                             .cast(TenantAsset.class)
                             .map(TenantAsset::getTenantId)
                             .distinct()
                             .collectList(),
                        //成员
                        cache
                            .filter(binding -> binding instanceof TenantAsset)
                            .cast(TenantAsset.class)
                            .map(TenantAsset::getOwnerId)
                            .distinct()
                            .collectList(),
                        //维度绑定信息
                        cache
                            .map(AssetsUtils::bindingToMap)
                            .distinct()
                            .collectList()
                    )
                    //设置到配置中
                    .flatMap(tp4 -> tp4
                        .getT1()
                        .setConfigs(
                            PropertyConstants.tenantId.value(tp4.getT2()),
                            PropertyConstants.tenantMemberId.value(tp4.getT3()),
                            PropertyConstants.bindings.value(tp4.getT4())
                        )
                    );
            })
            .then(Flux
                      .fromIterable(noBindDevice)
                      .flatMap(deviceRegistry::getDevice)
                      .flatMap(device -> device
                          .removeConfigs(PropertyConstants.tenantId,
                                         PropertyConstants.tenantMemberId,
                                         PropertyConstants.bindings))
                      .then()
            );
//        return assetManager
//            .getTenantAssets(DeviceAssetType.device, deviceIds)
//            .groupBy(TenantAsset::getAssetId, Integer.MAX_VALUE)
//            .flatMap(group -> {
//                String deviceId = group.key();
//                noBindDevice.remove(deviceId);
//                Flux<TenantAsset> cache = group.cache();
//                return Mono
//                    .zip(
//                        //设备
//                        deviceRegistry.getDevice(deviceId),
//                        //租户
//                        cache.map(TenantAsset::getTenantId).distinct().collectList(),
//                        //成员
//                        cache.map(TenantAsset::getOwnerId).distinct().collectList()
//                    )
//                    //设置到配置中
//                    .flatMap(tp2 -> tp2.getT1()
//                                       .setConfigs(PropertyConstants.tenantId.value(tp2.getT2()),
//                                                   PropertyConstants.tenantMemberId.value(tp2.getT3())));
//            })
//            .then(Flux
//                      .fromIterable(noBindDevice)
//                      .flatMap(deviceRegistry::getDevice)
//                      .flatMap(device -> device
//                          .removeConfigs(PropertyConstants.tenantId,
//                                         PropertyConstants.tenantMemberId))
//                      .then()
//            );
    }

}
